package com.atguigu.hbase;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * Created by Smexy on 2023/1/10
 *
 *
 *  铁律:  客户端连接hbase，只需要知道hbase存储元数据所用的zk集群的地址即可！
 *
 *          类似： 执行HQL时，必须先连接mysql查询元数据，才能知道要读写表数据所在的hdfs路径！
 *
 *  ①创建客户端连接
 *          ConnectionFactory.
 *              谁创建，谁关闭
 *          Connection: 可以用来去连接集群的服务(hmaster, hregionserver)，维护元数据缓存，连接zookeeper，这些信息都可以被
 *                  这个连接中所创建的Table 及 Admin 共享。
 *                          Table:  表对象，负责DML ，对数据的增删改查
 *                          Admin:  管理员，负责DDL， 建表，删表，改表等操作
 *
 *          创建连接的注意事项:
 *                  Connection的创建是重量级(消耗资源和时间)的，而且Connection是线程安全的，意味着，只需在客户端App中创建一次。
 *                  声明为成员变量，在不同的线程中共享同一个连接。
 *
 *          Table和Admin的创建都是轻量级，不是线程安全的。不建议共享，池化，缓存Table和Admin。
 *
 *
 *  ②连接服务端
 *  ③准备命令
 *      读：  Table.get(Get a)
 *           Table.getScanner(Scan s)
 *      写：  Table.put(Put a)
 *           Table.delete(Detele a)
 *
 *      工具类:
 *              Bytes:   将数据类型和 byte[]相互转换。
 *                      把常见的数据类型转为 byte[]:  Bytes.toBytes(T t)
 *                      把byte[]转换为常见的类型:  Bytes.toXxx(byte [] t)
 *
 *              CellUtil: 用来获取某个Cell中的某个属性。
 *                      获取Rowkey:  CellUtil.cloneRowkey(Cell c)
 *                      获取xxx:   CellUtil.cloneXxx(Cell c)
 *  ④执行命令
 *  ⑤读操作，解析返回值
 *  ⑥关闭连接
 */
public class HBaseUtil
{
    //一个App中只需要创建一个Connection(单例)
   private static Connection connection;

   //遍历一行Result
    public static void parseResult(Result result){

        //获取这行的所有单元格
        Cell[] cells = result.rawCells();

        for (Cell cell : cells) {

            //获取其中的属性
            System.out.println("rowkey=>" + Bytes.toString(CellUtil.cloneRow(cell)) + "," +
                Bytes.toString(CellUtil.cloneFamily(cell)) + ":" +
                Bytes.toString(CellUtil.cloneQualifier(cell)) + "=>" +
                Bytes.toString(CellUtil.cloneValue(cell)) + ",ts=>" +
                cell.getTimestamp());

        }

    }

   public static Put createPut(String rowkey,String cf,String cq,String value){

       Put put = new Put(Bytes.toBytes(rowkey));
       //向这一行，添加一个Cell
       put.addColumn(Bytes.toBytes(cf),Bytes.toBytes(cq),Bytes.toBytes(value));

       return put;

   }

   //传入表名，获取Table对象
   public static Table getTable(String table) throws IOException {

       if (StringUtils.isBlank(table)){
           throw new RuntimeException("表名非法!");
       }

       Table t = connection.getTable(TableName.valueOf(table));

       return t;

   }

   //随着类的加载而执行，只执行一次
   static {
       try {
           //这行代码运行完后，会自动读取类路径下 hadoop的配置文件和hbase的配置文件
           connection = ConnectionFactory.createConnection();
       } catch (IOException e) {
           e.printStackTrace();
       }
   }

   public static void close() throws IOException {
       if (connection != null){
           connection.close();
       }
   }

}
